package edu.uab.enronemail.helper;

import java.io.IOException;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.james.mime4j.MimeException;
import org.apache.james.mime4j.stream.EntityState;
import org.apache.james.mime4j.stream.MimeConfig;
import org.apache.james.mime4j.stream.MimeTokenStream;
import org.apache.tools.ant.filters.StringInputStream;

import edu.uab.enronemail.request.EmailDocument;

public class SequenceFileReader {
	
	static SequenceFile.Reader reader = null;
	static MimeTokenStream mimeStream;
	static MimeConfig mimeConfig = new MimeConfig();
	
	public static int errorCount = 0;
	
	public static void configure(){
		mimeConfig.setMaxHeaderLen(-1);
		mimeConfig.setMaxLineLen(-1);
		mimeStream= new MimeTokenStream(mimeConfig);
		
		Configuration conf = new Configuration();
		FileSystem fs;
		try {
			fs = FileSystem.get( URI.create("/home/spoudel/email-search-indexing/email-search-indexing/chunk-0"), conf);
			Path path = new Path("/home/spoudel/email-search-indexing/email-search-indexing/chunk-0");
			reader = new SequenceFile.Reader(fs, path, conf);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}

	public EmailDocument readNext(){
		try {
			if(reader==null) configure();
			Text key = new Text();
			Text value = new Text();
			if(!reader.next(key,value)) return null;
			return parseEmail(value.toString(), mimeStream);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}
	
	public static EmailDocument parseEmail(String text, MimeTokenStream mimeStream){
		mimeStream.parse(new StringInputStream(text));
		EmailDocument emailDoc = new EmailDocument();
		try {
			for(EntityState state = mimeStream.getState(); state!=EntityState.T_END_OF_STREAM;state = mimeStream.next()){
				if(EntityState.T_FIELD.equals(state)){
					emailDoc.addParts(mimeStream.getField().getName(), mimeStream.getField().getBody());
				}
				else if(EntityState.T_BODY.equals(state)){
					emailDoc.setEmailBody(mimeStream.getInputStream().toString());
					emailDoc.setEmailBodyTokens(emailDoc.getEmailBody());
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		} catch (MimeException e) {
			e.printStackTrace();
			//errorCount++;
		}
		return emailDoc;
	}
		
}
